package com.jiang.movie.demos

import com.jiang.movie.metrics.{BestFilmsByOverallRating, GenresByAverageRating, MostRatedFilms}
import org.apache.spark.SparkContext
import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}

object DemoMainApp {

  // 文件路径
  private val MOVIES_CSV_FILE_PATH = "file:///E:/Study_work/Bigdata/Spark/data/ml-25m/movies.csv"
  private val RATINGS_CSV_FILE_PATH = "file:///E:/Study_work/Bigdata/Spark/data/ml-25m/ratings.csv"

  def main(args: Array[String]): Unit = {

    //TODO 0.准备环境
    val spark: SparkSession = SparkSession.builder().appName("sparksql").master("local[*]").getOrCreate()
    val ssc: SparkContext = spark.sparkContext
    ssc.setLogLevel("WARN")
    import spark.implicits._
    import org.apache.spark.sql.functions._
    // schema信息
    val schemaLoader = new SchemaLoader
    // 读取Movie数据集
    val movieDF: DataFrame = spark.read.format("csv")
      .option("header", "true")
      .load(MOVIES_CSV_FILE_PATH)

    // 读取Rating数据集
//    val ratingDF: DataFrame = spark.read.csv(RATINGS_CSV_FILE_PATH)
    val ratingDF: DataFrame = spark.read.format("csv")
      .option("header", "true")
      .load(RATINGS_CSV_FILE_PATH)

    // 需求1：查找电影评分个数超过5000,且平均评分较高的前十部电影名称及其对应的平均评分
    val bestFilmsByOverallRating = new BestFilmsByOverallRating
    bestFilmsByOverallRating.run(movieDF,ratingDF,spark)

    // 需求2：查找每个电影类别及其对应的平均评分
    val genresByAverageRating = new GenresByAverageRating
    genresByAverageRating.run(movieDF, ratingDF, spark)

    // 需求3：查找被评分次数较多的前十部电影
    val mostRatedFilms = new MostRatedFilms
    mostRatedFilms.run(movieDF, ratingDF, spark)

      spark.stop()
  }

}
